package com.siao.net.socket;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.StandardSocketOptions;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.nio.channels.ShutdownChannelGroupException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ConnectionPool extends ConnectionService {

	private static final Logger logger = LoggerFactory.getLogger(ConnectionPool.class);

	/**
	 * 连接数
	 */
	private final int size;

	/**
	 * 获取连接数
	 * @return
	 */
	public int getSize() {
		return this.size;
	}

	/**
	 * 当前连接状态
	 */
	private boolean connected;

	/**
	 * 获取当前连接状态
	 * @return
	 */
	public boolean isConnected() {
		return this.connected;
	}

	/**
	 * 连接地址
	 */
	private InetSocketAddress address;

	/**
	 * 获取连接地址
	 * @return
	 */
	public InetSocketAddress getAddress() {
		return this.address;
	}

	/**
	 * 线程数
	 */
	private final int threadCount;

	/**
	 * 空闲连接
	 */
	private final ConcurrentLinkedQueue<Connection> idleConnections;

	/**
	 * 忙碌连接
	 */
	private final ConcurrentHashMap<Connection, Object> busyConnections;

	/**
	 * Socket通道组
	 */
	private AsynchronousChannelGroup channelGroup;

	/**
	 * 构造函数
	 */
	public ConnectionPool() {
		this(Runtime.getRuntime().availableProcessors());
	}

	/**
	 * 构造函数
	 * @param size
	 */
	public ConnectionPool(int size) {
		this(size, Runtime.getRuntime().availableProcessors());
	}

	/**
	 * 构造函数
	 * @param size
	 * @param threadCount
	 */
	public ConnectionPool(int size, int threadCount) {
		this.size = size;
		this.threadCount = threadCount;
		this.idleConnections = new ConcurrentLinkedQueue<Connection>();
		this.busyConnections = new ConcurrentHashMap<Connection, Object>();
	}

	public String toString() {
		return ("idle:" + this.idleConnections.size() + ",busy:" + this.busyConnections.size());
	}

	/**
	 * 连接
	 * @param address
	 */
	public synchronized void connect(InetSocketAddress address) {
		if (this.connected) {
			this.close();
		}

		this.connected = true;

		try {
			this.address = address;
			this.channelGroup = AsynchronousChannelGroup.withFixedThreadPool(this.threadCount, Executors.defaultThreadFactory());

			for (int i = 0; i < this.size; i++) {
				this.idleConnections.add(this.createConnection());
			}
		} catch (IOException e) {
			logger.error("ConnectionPool have failed to connect: " + this.address + "," + e.getMessage());
		}
	}

	/**
	 * 请求数据
	 * @param data
	 * @return
	 */
	public Object request(IDataPack data) {
		return this.request(data, 3000);
	}

	/**
	 * 发送数据请求，并等待返回数据
	 * @param data
	 * @param timeout
	 * @return
	 */
	public Object request(IDataPack data, int timeout) {
		synchronized (this) {
			if (!this.connected) {
				return null;
			}
		}

		Connection connection = this.getIdleConnection();
		connection.send(data);

		this.busyConnections.put(connection, data);

		synchronized (connection) {
			try {
				connection.wait(timeout);
			} catch (InterruptedException e) {
			}
		}

		Object result = this.busyConnections.get(connection);
		this.recoverConnection(connection);

		return result;
	}

	/**
	 * 异步发送数据请求
	 * @param address
	 * @param data
	 */
	public void post(IDataPack data) {
		synchronized (this) {
			if (!this.connected) {
				return;
			}
		}

		Connection connection = this.getIdleConnection();
		this.busyConnections.put(connection, data);
		connection.send(data);
		this.recoverConnection(connection);
	}

	/**
	 * 关闭
	 */
	public synchronized void close() {
		this.connected = false;

		while (!this.busyConnections.isEmpty()) {
		}

		this.busyConnections.clear();

		while (!this.idleConnections.isEmpty()) {
			this.idleConnections.poll().close();
		}

		this.idleConnections.clear();

		if (this.channelGroup != null) {
			if (!this.channelGroup.isShutdown()) {
				this.channelGroup.shutdown();
			}
			this.channelGroup = null;
		}
	}

	/**
	 * 创建一个连接
	 * @return
	 */
	private Connection createConnection() {
		Connection connection = new Connection(this);

		try {
			AsynchronousSocketChannel channel = AsynchronousSocketChannel.open(this.channelGroup);
			channel.setOption(StandardSocketOptions.TCP_NODELAY, true);
			channel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
			channel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);
			channel.connect(this.address, connection, new ConnectionAcceptHandler());

			synchronized (connection) {
				connection.wait(3000);
			}

			if (channel.isOpen()) {
				connection.register(channel);
			}
		} catch (InterruptedException e) {
		} catch (IOException e) {
			logger.error("Create Connection Failed");
		}

		return connection;
	}

	/**
	 * 获取一个空闲连接
	 * @return
	 */
	private Connection getIdleConnection() {
		int times = 0;
		Connection connection = this.idleConnections.poll();
		while ((connection == null || !connection.isConnected()) && times < 3) {
			connection = this.createConnection();
			times++;
		}
		return connection;
	}

	/**
	 * 回收一个连接
	 * @param connection
	 */
	private void recoverConnection(Connection connection) {
		if (!this.busyConnections.containsKey(connection)) {
			return;
		}

		this.busyConnections.remove(connection);

		if (this.idleConnections.size() < this.size) {
			this.idleConnections.add(connection);
		} else {
			connection.close();
		}
	}

	/**
	 * @author SiaoLeon
	 * 连接处理
	 */
	class ConnectionAcceptHandler implements CompletionHandler<Void, Connection> {

		@Override
		public void completed(Void result, Connection connection) {
			synchronized (connection) {
				connection.notify();
			}
		}

		@Override
		public void failed(Throwable exc, Connection connection) {
			if (exc instanceof ShutdownChannelGroupException) {
				logger.error("ShutdownChannelGroupException");
				return;
			}

			ConnectionPool.this.connectionFailed(connection);

			exc.printStackTrace();
			logger.error(exc.getMessage());

			synchronized (connection) {
				connection.notify();
			}
		}

	}

	@Override
	protected void connectionRegister(Connection connection) {
		super.connectionRegister(connection);
	}

	@Override
	protected void connectionFailed(Connection connection) {
		super.connectionFailed(connection);
	}

	@Override
	protected void connectionSend(Connection connection, IDataPack data) {
		super.connectionSend(connection, data);
	}

	@Override
	protected void connectionReceive(Connection connection, Object data) {
		super.connectionReceive(connection, data);

		this.busyConnections.put(connection, data);

		synchronized (connection) {
			connection.notify();
		}
	}

	@Override
	protected void connectionClose(Connection connection) {
		super.connectionClose(connection);
	}

	@Override
	protected void connectionDisconnect(Connection connection) {
		super.connectionDisconnect(connection);
	}

}
